package com.yanfan.iot.service.impl;

import java.sql.SQLException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson2.JSON;
import com.dtflys.forest.http.ForestRequest;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.http.service.HttpClientFactory;
import com.yanfan.http.service.SuccessCondition;
import com.yanfan.iot.domain.HttpClient;
import com.yanfan.iot.domain.MqttClient;
import com.yanfan.iot.domain.MultipleDataSource;
import com.yanfan.iot.ruleEngine.DataSourceService;
import com.yanfan.iot.ruleEngine.MqttClientFactory;
import com.yanfan.iot.service.IHttpClientService;
import com.yanfan.iot.service.IMqttClientService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.yanfan.iot.mapper.BridgeMapper;
import com.yanfan.iot.domain.Bridge;
import com.yanfan.iot.service.IBridgeService;

import javax.annotation.PostConstruct;

/**
 * 数据桥接Service业务层处理
 *
 * @author zhuangpeng.li
 * @date 2024-06-06
 */
@Service
@Slf4j
public class BridgeServiceImpl implements IBridgeService {
    @Autowired
    private BridgeMapper bridgeMapper;

    @Autowired
    private IHttpClientService httpClientService;

    @Autowired
    private IMqttClientService mqttClientService;

    @Autowired
    private DataSourceService dataSourceService;

    @PostConstruct
    public void initBridge() {
        List<Bridge> list = bridgeMapper.selectBridgeList(new Bridge());
        for (Bridge bridge : list) {
            if (bridge.getType() == 4) {
                MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                if (config != null) {
                    try {
                        MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config));
                        if (client != null && bridge.getDirection() == 1 && !Objects.equals(bridge.getRoute(), "")) {
                            MqttClientFactory.addSubscribe(client, bridge.getRoute());
                        }
                    } catch (MqttException e) {
                        throw new RuntimeException(e);
                    }

                }
            }
        }
    }

    /**
     * 查询数据桥接
     *
     * @param id 数据桥接主键
     * @return 数据桥接
     */
    @Override
    public Bridge selectBridgeById(Long id) {
        return bridgeMapper.selectBridgeById(id);
    }

    /**
     * 查询数据桥接列表
     *
     * @param bridge 数据桥接
     * @return 数据桥接
     */
    @Override
    public List<Bridge> selectBridgeList(Bridge bridge) {
        return bridgeMapper.selectBridgeList(bridge);
    }

    /**
     * 新增数据桥接
     *
     * @param bridge 数据桥接
     * @return 结果
     */
    @Override
    public int insertBridge(Bridge bridge) {
        if (bridge.getType() == 4) {
            MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
            if (config != null) {
                try {
                    MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config));
                    if (bridge.getConfigJson() != null && bridge.getDirection() == 1) {
                        bridge.setStatus("1");
                        MqttClientFactory.addSubscribe(client, bridge.getRoute());
                    }
                } catch (MqttException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        bridge.setCreateTime(DateUtils.getNowDate());
        return bridgeMapper.insertBridge(bridge);
    }

    /**
     * 修改数据桥接
     *
     * @param bridge 数据桥接
     * @return 结果
     */
    @Override
    public int updateBridge(Bridge bridge) {
        bridge.setUpdateTime(DateUtils.getNowDate());
        if (bridge.getType() == 4) {
            if (bridge.getConfigJson() != null && bridge.getDirection() == 1) {
                MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                bridge.setStatus("1");
                MqttClientFactory.addSubscribe(config.getName(), bridge.getRoute());
            }
        }
        return bridgeMapper.updateBridge(bridge);
    }

    @Override
    public int connect(Bridge bridge) {
        AtomicInteger result = new AtomicInteger();
        if (bridge.getType() == 3) {
            HttpClient config = JSON.parseObject(bridge.getConfigJson(), HttpClient.class);
            ForestRequest request = HttpClientFactory.instance(httpClientService.buildhttpclientconfig(config));
            request.successWhen(SuccessCondition.class).onSuccess(((data, req, res) -> {
                bridge.setStatus("1");
                result.set(1);
            })).onError((ex, req, res) -> {
                bridge.setStatus("0");
                result.set(0);
            }).execute();
        } else if (bridge.getType() == 4) {
            MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
            if (config != null) {
                try {
                    MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config));
                    if (client != null && client.isConnected()) {
                        bridge.setStatus("1");
                        result.set(1);
                    } else {
                        bridge.setStatus("0");
                        result.set(0);
                    }
                } catch (MqttException e) {
                    log.error("=>mqtt客户端创建错误");
                }
            }
        } else if (bridge.getType() == 5) {
            MultipleDataSource config = JSON.parseObject(bridge.getConfigJson(), MultipleDataSource.class);
            DruidDataSource dataSource = dataSourceService.createDataSource(config);
            try {
                if (dataSource != null && dataSource.getConnection() != null) {
                    bridge.setStatus("1");
                    result.set(1);
                } else {
                    bridge.setStatus("0");
                    result.set(0);
                }
            } catch (SQLException e) {
                log.error("=>数据源连接错误");
            }
        }
        bridgeMapper.updateBridge(bridge);
        return result.get();
    }

    /**
     * 批量删除数据桥接
     *
     * @param ids 需要删除的数据桥接主键
     * @return 结果
     */
    @Override
    public int deleteBridgeByIds(Long[] ids) {
        return bridgeMapper.deleteBridgeByIds(ids);
    }

    /**
     * 删除数据桥接信息
     *
     * @param id 数据桥接主键
     * @return 结果
     */
    @Override
    public int deleteBridgeById(Long id) {
        return bridgeMapper.deleteBridgeById(id);
    }
}
